%%%-------------------------------------------------------------------
%%% File    : mapred_test.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created :  4 Aug 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(mapred_tests).

-include_lib("eunit/include/eunit.hrl").

%% API
-export([]).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Brute & lookup search of: 
%%	{Id,{"Roberts","Fleming"},IdF,{FrName,FrFamily}} -> PIPE	LOG  
%%--------------------------------------------------------------------
person_friends_1_test()->
	MapByS=fun({spog,S,P,O,G}=Item) ->
				   {S,Item}
		   end,
	MapStrByS=fun({spog,S,P,{'$lit',Str,string}=O,G}=Item) ->
					  {S,Str}
			  end,
	St=now(),
	Spec={red, 
		  fun(S,[Iter1,Iter2])->
				  {_,FrIter}=rdf_mapred_builder:find(
							   [{s,S},{p,{onto,"hasFriend"}}]),
				  Iter3=iter_cache:new(FrIter),
				  iter_mapper:new(
					iter_join_cprod:all(
					  [Iter1, Iter2, Iter3]),
					fun([N,F,{spog,S,_P,O,_G}])->
							{S,N,F,O}
					end)
		  end, 
		  [{map,
			MapStrByS,   
			rdf_mapred_builder:find([{p,{onto,"hasFamilyName"}},{o,{'$lit',"Roberts",string}}])
		   },
		   {map,
			MapStrByS,
			rdf_mapred_builder:find([{p,{onto,"hasName"}},{o,{'$lit',"Fleming",string}}])
		   }
		  ]},
	Spec1={red,
		   fun(O,[Iter1])->
				   {_,I}=rdf_mapred_builder:find([{s,O}]),
				   BatchIter=iter_cache:new(I),
				   FrFIter=iter_filter:new(
							 BatchIter, 
							 fun(Spog)when element(3,Spog)=={onto,"hasFamilyName"} ->
									 true;
								(_) ->
									 false
							 end),
				   FrNIter=iter_filter:new(
							 BatchIter, 
							 fun(Spog)when element(3,Spog)=={onto,"hasName"} ->
									 true;
								(_) ->
									 false
							 end),
				   iter_mapper:new(
					 iter_join_cprod:all(
					   [Iter1, FrNIter, FrFIter]),
					 fun([{S,N,F},
						  {spog,S1,_P1,{'$lit',FN,string},_G1},
						  {spog,S1,_P2,{'$lit',FF,string},_G2}])->
							 {S,{N,F},S1,{FN,FF}}
					 end)
		   end
			,
			[{map,
			 fun({S,N,F,O})->
					 {O,{S,N,F}}
			 end,
			 Spec
			}]},
	Iter=iter_counter:new_progress_observer(
		   mapred:iter(Spec1),1000,
		   fun(Prog,Count)->
				   io:format("~p: ~p~n",[Prog,Count])
		   end),
	Pump=pump:new(Iter,pipe_zip:new(1000,pipe_disk_logger:new([{name,rob_fle}]))),
	pump:init(Pump),
	{St,now()}.
%%====================================================================
%% Internal functions
%%====================================================================


